package My

import (
	"errors"
	"fmt"
	"db2s/ETL"
	ea "db2s/encryptionAlgorithm"
	"db2s/global"
	"math"
	"strconv"
	"strings"
)

type TableInfoMeta struct {
}

func (my TableInfoMeta) AvgRowLength(s global.TablesMetaInfoInput) (res global.AvgRowLengthReturnResult, err error) {
	var (
		event = "[MySQLAvgRowLength]"
		f1    any
	)
	res.BaseResult.Sql = fmt.Sprintf("select `AVG_ROW_LENGTH` as `AvgRowLength` from information_schema.tables where table_schema='%v' and table_name ='%v'", s.TableInfo.Schema, s.TableInfo.Table)
	if res.BaseResult.TimeOut, f1, err = ExecSelectSql(ExecSqlInputS{
		Ss:           s,
		Event:        event,
		SelectSqlStr: res.BaseResult.Sql,
		Expect:       ETL.Map,
	}); err != nil || f1 == nil {
		if err != nil {
			err = errors.New(fmt.Sprintf("%v %v", event, err))
		}
		return
	}
	if l, ok := f1.(map[string]any)["AvgRowLength"]; ok {
		if res.Result, err = strconv.ParseInt(fmt.Sprintf("%v", l), 10, 64); err != nil {
			err = errors.New(fmt.Sprintf("%v %v", event, err))
			return
		}
	}
	return
}

/*
DataSize 查询表的统计信息，获取表的数据大小，数据大小+索引大小，单位MB
*/
func (my TableInfoMeta) DataSize(s global.TablesMetaInfoInput) (result global.DataSizeReturnResult, err error) {
	var (
		event = "[MySQLDataSize]"
		f1    any
	)
	result.Sql = fmt.Sprintf("SELECT ROUND(SUM(DATA_LENGTH) / 1048576, 2) AS `dataSizeMB`,ROUND(SUM(INDEX_LENGTH) / 1048576, 2) AS `indexSizeMB` FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '%v' AND TABLE_NAME = '%v'", s.TableInfo.Schema, s.TableInfo.BackendTableName)
	if result.ExecTimeout, f1, err = ExecSelectSql(ExecSqlInputS{
		Ss:           s,
		Event:        event,
		SelectSqlStr: result.Sql,
		Expect:       ETL.Map,
	}); err != nil || f1 == nil {
		if err != nil {
			err = errors.New(fmt.Sprintf("%v %v", event, err))
		}
		return
	}
	for k, v := range []string{"dataSizeMB", "indexSizeMB"} {
		if _, ok := f1.(map[string]interface{})[v]; !ok {
			err = errors.New(fmt.Sprintf("%v query result not include %v. process:{sql:%v,result:%v}", v, event, result.Sql, f1))
			return
		}
		v1 := fmt.Sprintf("%v", f1.(map[string]interface{})[v])
		if strings.HasPrefix(v1, ".") {
			v1 = fmt.Sprintf("0%v", v1)
		}
		if !isNumeric(v1) {
			err = errors.New(fmt.Sprintf("%v The result contains illegal non-numeric data. process:{sql:%v,result:%v}", event, result.Sql, v1))
			return
		}
		var v2 float64
		if v2, err = strconv.ParseFloat(v1, 64); err != nil {
			err = errors.New(fmt.Sprintf("%v string strconv float64 fail, strconv data is {%v} error is {%v}", event, v1, err))
			return
		}
		switch k {
		case 0:
			result.DataSizeMB = math.Round(v2*100) / 100
		case 1:
			result.IndexSizeMB = math.Round(v2*100) / 100
		}
	}
	return
}
func (my TableInfoMeta) CharacterCollate(s global.TablesMetaInfoInput, collate string) (result string, err error) {
	var (
		event = "[MySQLComment]"
		f1    any
	)
	strSql := fmt.Sprintf("select CHARACTER_SET_NAME from INFORMATION_SCHEMA.COLLATIONS WHERE COLLATION_NAME='%v'", collate)
	if _, f1, err = ExecSelectSql(ExecSqlInputS{
		Ss:           s,
		Event:        event,
		SelectSqlStr: strSql,
		Expect:       ETL.String,
	}); err != nil || f1 == nil {
		if err != nil {
			err = errors.New(fmt.Sprintf("%v %v", event, err))
		}
		return
	}
	switch {
	case strings.EqualFold(fmt.Sprintf("%v", f1), "<null>"), strings.EqualFold(fmt.Sprintf("%v", f1), "<entry>"), len(fmt.Sprintf("%v", f1)) == 0:
		result = "<entry>"
	default:
		result = fmt.Sprintf("%v", f1)
	}
	return
}

// Comment 获取表的comment信息
func (my TableInfoMeta) Comment(s global.TablesMetaInfoInput) (result global.TableCommentReturnResult, err error) {
	var (
		event = "[MySQLComment]"
		f1    any
	)
	strSql := fmt.Sprintf("SELECT REGEXP_SUBSTR(TABLE_COLLATION, '^[^_]+') AS `character`,TABLE_COLLATION as `collation`,TABLE_COMMENT as `comment` FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '%v' AND TABLE_NAME='%v'", s.TableInfo.Schema, s.TableInfo.Table)
	if result.BaseResult.TimeOut, f1, err = ExecSelectSql(ExecSqlInputS{
		Ss:           s,
		Event:        event,
		SelectSqlStr: strSql,
		Expect:       ETL.Map,
	}); err != nil || f1 == nil {
		if err != nil {
			err = errors.New(fmt.Sprintf("%v %v", event, err))
		}
		return
	}
	if v, ok := f1.(map[string]any)["comment"]; ok {
		switch {
		case strings.EqualFold(fmt.Sprintf("%v", v), "<null>"), strings.EqualFold(fmt.Sprintf("%v", v), "<entry>"), len(fmt.Sprintf("%v", v)) == 0:
			result.Result.Comment = "<entry>"
		default:
			result.Result.Comment = fmt.Sprintf("%v", v)
		}
	}
	if v, ok := f1.(map[string]any)["collation"]; ok {
		switch {
		case strings.EqualFold(fmt.Sprintf("%v", v), "<null>"), strings.EqualFold(fmt.Sprintf("%v", v), "<entry>"), len(fmt.Sprintf("%v", v)) == 0:
			result.Result.Collate = "<entry>"
		default:
			result.Result.Collate = fmt.Sprintf("%v", v)
		}
	}
	if v, ok := f1.(map[string]any)["character"]; ok {
		switch {
		case strings.EqualFold(fmt.Sprintf("%v", v), "<null>"), strings.EqualFold(fmt.Sprintf("%v", v), "<entry>"), len(fmt.Sprintf("%v", v)) == 0:
			result.Result.Collate = "<entry>"
		default:
			result.Result.Character = fmt.Sprintf("%v", v)
		}
	}
	return
}

// IfTableLob 判断表是否存在lob字段
func (my TableInfoMeta) IfTableLob(s global.TablesMetaInfoInput) (result global.TableBigVarcharMeta, err error) {
	var (
		event                  = "[MySQLIfTableLob]"
		columnName, columnType []string
		res                    any
	)
	result.BaseResult.Sql = fmt.Sprintf("SELECT `COLUMN_NAME` as `columnName`, `DATA_TYPE` as `columnType` FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = '%v'  AND TABLE_NAME = '%v' AND DATA_TYPE IN ('tinyblob', 'blob', 'mediumblob', 'longblob')", s.TableInfo.Schema, s.TableInfo.BackendTableName)
	if result.BaseResult.TimeOut, res, err = ExecSelectSql(ExecSqlInputS{
		Ss:           s,
		Event:        event,
		SelectSqlStr: result.BaseResult.Sql,
		Expect:       ETL.SMap,
	}); err != nil || res == nil {
		if err != nil {
			err = errors.New(fmt.Sprintf("%v %v", event, err))
		}
		return
	}
	for _, v := range res.([]map[string]any) {
		columnName = append(columnName, fmt.Sprintf("%v", v["columnName"]))
		columnType = append(columnType, fmt.Sprintf("%v", v["columnType"]))
	}
	if len(columnName) > 0 && len(columnType) > 0 {
		result = global.TableBigVarcharMeta{
			ColumnName: columnName,
			ColumnType: columnType,
		}
	}
	return
}

// IfBigVarchar 判断表是否存在varchar>1000
func (my TableInfoMeta) IfBigVarchar(s global.TablesMetaInfoInput) (result global.TableBigVarcharMeta, err error) {
	var (
		event                  = "[MySQLIfBigVarchar]"
		columnName, columnType []string
		res                    any
	)
	result.BaseResult.Sql = fmt.Sprintf("SELECT `COLUMN_NAME` as `columnName`, `DATA_TYPE` as `columnType`,`CHARACTER_MAXIMUM_LENGTH` as `dataLength` FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = '%v' AND TABLE_NAME = '%v' AND DATA_TYPE in ('varchar','text','mediumtext','longtext')  AND CHARACTER_MAXIMUM_LENGTH > %d", s.TableInfo.Schema, s.TableInfo.BackendTableName, 1000)
	if result.BaseResult.TimeOut, res, err = ExecSelectSql(ExecSqlInputS{
		Ss:           s,
		Event:        event,
		SelectSqlStr: result.BaseResult.Sql,
		Expect:       ETL.SMap,
	}); err != nil || res == nil {
		if err != nil {
			err = errors.New(fmt.Sprintf("%v %v", event, err))
		}
		return
	}
	for _, v := range res.([]map[string]any) {
		columnName = append(columnName, fmt.Sprintf("%v", v["columnName"]))
		columnType = append(columnType, fmt.Sprintf("%v", v["columnType"]))
	}
	if len(columnName) > 0 && len(columnType) > 0 {
		result = global.TableBigVarcharMeta{
			ColumnName: columnName,
			ColumnType: columnType,
		}
	}
	return
}
func (my TableInfoMeta) TableDistributed(s global.TablesMetaInfoInput) (any, error) {
	var (
		tpc []global.GetTableDistributed
	)
	tpc = append(tpc, global.GetTableDistributed{
		Schema:           s.TableInfo.Schema,
		Table:            s.TableInfo.Table,
		DistributeMode:   "NORMAL",
		ShardName:        "single",
		BackendTableName: s.TableInfo.Table,
	})
	return tpc, nil
}

func (my TableInfoMeta) DropPartitionSql(s global.TablesMetaInfoInput) (any, error) {
	var alterGather []string
	for k := range s.PartitionMetaGarth.DstMeta {
		alterGather = append(alterGather, fmt.Sprintf("alter table `%v`.`%v` drop partition `%v`;", s.TableInfo.Schema, s.TableInfo.Table, k))
	}
	return global.AlterSqlResult{
		Sql: strings.Join(alterGather, "\n"),
	}, nil
}
func (my TableInfoMeta) AddPartitionSql(s global.TablesMetaInfoInput) (any, error) {
	var alterGather []string
	for _, v := range s.PartitionMetaGarth.SrcMeta {
		switch v.Meth {
		case "RANGE":
			alterGather = append(alterGather, fmt.Sprintf("alter table `%v`.`%v` add partition (partition `%v` VALUES LESS THAN (%v));", s.TableInfo.Schema, s.TableInfo.Table, v.Name, v.Rules))
		case "LIST":
			alterGather = append(alterGather, fmt.Sprintf("alter table `%v`.`%v` add partition (partition `%v` values in (%v));", s.TableInfo.Schema, s.TableInfo.Table, v.Name, v.Rules))
		}
	}
	return global.AlterSqlResult{
		Sql: strings.Join(alterGather, "\n"),
	}, nil
}
func (my TableInfoMeta) ModifyPartitionSql(s global.TablesMetaInfoInput) (any, error) {
	var alterGather []string
	for k := range s.PartitionMetaGarth.DstMeta {
		alterGather = append(alterGather, fmt.Sprintf("alter table `%v`.`%v` drop partition `%v`;", s.TableInfo.Schema, s.TableInfo.Table, k))
	}
	for _, v := range s.PartitionMetaGarth.SrcMeta {
		switch v.Meth {
		case "RANGE":
			alterGather = append(alterGather, fmt.Sprintf("alter table `%v`.`%v` add partition (partition `%v` VALUES LESS THAN (%v));", s.TableInfo.Schema, s.TableInfo.Table, v.Name, v.Rules))
		case "LIST":
			alterGather = append(alterGather, fmt.Sprintf("alter table `%v`.`%v` add partition (partition `%v` values in (%v));", s.TableInfo.Schema, s.TableInfo.Table, v.Name, v.Rules))
		}
	}
	return global.AlterSqlResult{
		Sql: strings.Join(alterGather, "\n"),
	}, nil
}

func (my TableInfoMeta) Distribution(s global.TablesMetaInfoInput) (any, error) {
	var (
		event       = "[TableDistribution]"
		execTimeout int64
		pp          []global.TableDistribution
	)
	vlog = fmt.Sprintf("(%d) %v Start querying the distribution of table %v.%v", s.LogSeq, event, s.TableInfo.Schema, s.TableInfo.Table)
	WLog.Debug(vlog)
	pp = append(pp, global.TableDistribution{
		Name:      fmt.Sprintf("%v.%v", s.TableInfo.Schema, s.TableInfo.Table),
		ShardMode: "NORMAL",
		ShardName: "single",
	})
	vlog = fmt.Sprintf("(%d) %v A sql statement was executed successfully. sql is {%v}, exec timeout is {%vms}", s.LogSeq, event, "no sql", execTimeout)
	WLog.Debug(vlog)
	vlog = fmt.Sprintf("(%d) %v The distribution query of table %v.%v is completed. result is {%v}", s.LogSeq, event, s.TableInfo.Schema, s.TableInfo.Table, pp)
	WLog.Debug(vlog)
	return pp, nil
}
func (my TableInfoMeta) DropIndexSql(s global.TablesMetaInfoInput) (any, error) {
	var (
		sqlS string
		//Event = "[DropIndexSql]"
	)
	switch s.IndexMetaGather.Dst.Type {
	case "pri":
		sqlS = fmt.Sprintf("alter table `%v`.`%v` drop primary key;", s.TableInfo.Schema, s.TableInfo.Table)
	case "uni", "mul":
		sqlS = fmt.Sprintf("alter table `%v.`%v drop index `%v`;", s.TableInfo.Schema, s.TableInfo.Table, s.IndexMetaGather.Dst.Name)
	}
	return global.AlterSqlResult{
		Sql: sqlS,
	}, nil
}
func (my TableInfoMeta) AddIndexSql(s global.TablesMetaInfoInput) (any, error) {
	var (
		sqlS             []string
		indexS           []string
		record, alterSql string
		//Event            = "[AddIndexSql]"
	)
	for _, v := range s.IndexMetaGather.Src.IndexColumn {
		indexS = append(indexS, fmt.Sprintf("%v", v.ColumnName))
	}
	switch s.IndexMetaGather.Src.Type {
	case "pri":
		if len(s.IndexMetaGather.PartColumn) > 0 {
			indexS = append(indexS, s.IndexMetaGather.PartColumn...)
		}
		record = fmt.Sprintf("primary key must include partition column")
		sqlS = append(sqlS, fmt.Sprintf("alter table `%v`.`%v` add primary key(`%v`);", s.TableInfo.Schema, s.TableInfo.Table, strings.Join(indexS, "`,`")))
	case "uni":
		if len(s.IndexMetaGather.PartColumn) > 0 {
			indexS = append(indexS, s.IndexMetaGather.PartColumn...)
		}
		record = fmt.Sprintf("unique index must include partition column")
		sqlS = append(sqlS, fmt.Sprintf("alter table `%v`.`%v` add unique index %v(`%v`);", s.TableInfo.Schema, s.TableInfo.Table, s.IndexMetaGather.Src.Name, strings.Join(indexS, "`,`")))
	case "mul":
		sqlS = append(sqlS, fmt.Sprintf("alter table `%v`.`%v` add index %v(`%v`);", s.TableInfo.Schema, s.TableInfo.Table, s.IndexMetaGather.Src.Name, strings.Join(indexS, "`,`")))
	}
	alterSql = strings.Join(sqlS, ",")
	return global.AlterSqlResult{
		Record: record,
		Sql:    alterSql,
	}, nil
}
func (my TableInfoMeta) ModifyIndexSql(s global.TablesMetaInfoInput) (any, error) {
	var (
		sqlS             []string
		indexS, triggerS []string
		//Event            = "[ModifyIndexSql]"
		alterSql, record string
		normal           bool
	)
	for _, v := range s.IndexMetaGather.Src.IndexColumn {
		indexS = append(indexS, fmt.Sprintf("%v", v.ColumnName))
	}
	for _, v := range s.IndexMetaGather.Dst.IndexColumn {
		triggerS = append(triggerS, fmt.Sprintf("%v", v.ColumnName))
	}
	//处理drop index
	switch s.IndexMetaGather.Dst.Type {
	case "pri":
		sqlS = append(sqlS, fmt.Sprintf("alter table `%v`.`%v` drop primary key", s.TableInfo.Schema, s.TableInfo.Table))
	case "uni", "mul":
		sqlS = append(sqlS, fmt.Sprintf("alter table `%v`.`%v` drop index `%v`", s.TableInfo.Schema, s.TableInfo.Table, s.IndexMetaGather.Dst.Name))
	}
	switch s.IndexMetaGather.Src.Type {
	case "pri":
		sqlS = append(sqlS, fmt.Sprintf(" add primary key(`%v`)", strings.Join(indexS, "`,`")))
		a, b := ea.CheckSum().Arrcmp(append(indexS, s.IndexMetaGather.PartColumn...), triggerS)
		if len(a) == 0 && len(b) == 0 {
			normal = true
			record = fmt.Sprintf("primary key must include partition column")
		}
	case "uni":
		sqlS = append(sqlS, fmt.Sprintf(" add unique index `%v`(`%v`)", s.IndexMetaGather.Src.Name, strings.Join(indexS, "`,`")))
		a, b := ea.CheckSum().Arrcmp(append(indexS, s.IndexMetaGather.PartColumn...), triggerS)
		if len(a) == 0 && len(b) == 0 {
			normal = true
			record = fmt.Sprintf("unique index must include partition column")
		}
	case "mul":
		sqlS = append(sqlS, fmt.Sprintf(" add index `%v`(`%v`)", s.IndexMetaGather.Src.Name, strings.Join(indexS, "`,`")))
	}
	if !normal {
		alterSql = strings.Join(sqlS, ",")
	}
	return global.AlterSqlResult{
		Record: record,
		Sql:    alterSql,
	}, nil
}
